package com.example.eeuse.monitor;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.example.eeuse.mapper.ee.TaskMonitorMapper;
import com.example.eeuse.model.ee.TaskMonitor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @author hly
 */
@Slf4j
@Service
public class QueueMonitorService {

    @Resource
    private TaskMonitorMapper taskMonitorMapper;

    @Resource
    private RedissonClient redissonClient;
    @PostConstruct
    public void init() throws InterruptedException {
        Thread.sleep(3000);
        // 启动队列消费任务
        startQueueMonitorWithRestart();
    }

    // 单线程执行器，用于提交并运行消费任务线程，确保一次只有一个消费线程在跑
    private static final ExecutorService executorService = Executors.newSingleThreadExecutor();

    // 单线程调度执行器，用于定期检查消费线程是否存活并重启（作为守护线程）
    private final ScheduledExecutorService monitorScheduler = Executors.newSingleThreadScheduledExecutor();

    // 标志位：用于标记当前是否已有消费线程在运行，防止重复提交
    private volatile boolean monitorRunning = false;

    /**
     * 启动队列监控并具备自动重启能力
     *
     * 技术要点：
     * - 使用 Redisson 的 RBlockingQueue 实现分布式阻塞队列消费
     * - 通过 ScheduledExecutorService 实现定时检查消费线程状态
     * - 使用 ExecutorService 保证消费任务单线程执行
     * - volatile + 定时器机制保证线程意外挂掉后可以被重新拉起
     * - 异常捕获防止线程中断退出，增强系统鲁棒性
     */
    public void startQueueMonitorWithRestart() {
        // 每隔 600 秒调度一次任务，检查消费线程是否在运行，若未运行则重新提交
        monitorScheduler.scheduleWithFixedDelay(() -> {
            if (!monitorRunning) {
                monitorRunning = true;

                // 提交消费线程到线程池中执行
                executorService.submit(() -> {
                    // 获取分布式阻塞队列对象（Redisson）
                    RBlockingQueue<JSONObject> blockingQueue = redissonClient.getBlockingQueue("iotChannelMonitorBlockingQueue");

                    while (true) {
                        try {
                            // 从队列中阻塞式获取一个元素（队列为空时阻塞）
                            JSONObject element = blockingQueue.take();

                            // 将 JSON 对象转换为业务对象 TaskMonitor
                            TaskMonitor taskMonitor = JSON.parseObject(element.toJSONString(), TaskMonitor.class);

                            // 打印日志，便于跟踪消费内容
                            log.info("Consumed:{}", JSONObject.toJSONString(taskMonitor));

                            // 写入数据库，记录消费信息
                            taskMonitorMapper.insert(taskMonitor);

                            // 其他业务逻辑可以追加在这里

                        } catch (InterruptedException e) {
                            // 捕获线程中断异常，并优雅退出循环
                            Thread.currentThread().interrupt();
                            log.error("Queue monitoring thread was interrupted.", e);
                            break;

                        } catch (Exception ex) {
                            // 捕获所有运行时异常，避免线程直接崩溃退出
                            log.error("Unexpected error while processing queue element", ex);
                            // 可选：某些严重异常也可以 break 出去，触发重启
                        }
                    }

                    // 当线程因异常或正常退出时，释放标志位，允许调度器重新启动线程
                    monitorRunning = false;
                });
            }
            // 初始立即执行，之后每 600 秒执行一次检查
        }, 0, 600, TimeUnit.SECONDS);
    }



}
